package com.usefullc.crawler.service;
import java.util.Date;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.usefullc.common.constants.DataConfigConstant;
import com.usefullc.common.enums.YesNoEnum;
import com.usefullc.common.util.JsonUtil;
import com.usefullc.common.util.http.HttpNewUtils;
import com.usefullc.common.util.http.ResponseRes;
import com.usefullc.common.util.secure.Base64;
import com.usefullc.crawler.constants.CrawlerCacheConstants;
import com.usefullc.crawler.constants.CrawlerConstants;
import com.usefullc.crawler.domain.*;
import com.usefullc.crawler.enums.ActiveInstanceStatusEnum;
import com.usefullc.crawler.enums.ProcessStatusEnum;
import com.usefullc.crawler.enums.RequestResultErrorStatusEnum;
import com.usefullc.crawler.param.CrawlerTaskSubmitParam;
import com.usefullc.crawler.query.ActiveQuery;
import com.usefullc.crawler.task.TaskMessage;
import com.usefullc.crawler.task.TaskMessageHandler;
import com.usefullc.crawler.task.TaskMessageQueue;
import com.usefullc.crawler.vo.CrawlerTaskVo;
import com.usefullc.crawler.vo.TaskTypeVo;
import com.usefullc.system.service.AbstractService;
import com.usefullc.system.service.DataConfigService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 爬虫任务服务类
 * @author: Sampson
 * @date: 2/19/23 5:36 PM
 */
@Service
public class CrawlerTaskService extends AbstractService {

    @Autowired
    private ActiveInstanceService activeInstanceService;

    @Autowired
    private ActiveSliceInstanceService activeSliceInstanceService;

    @Autowired
    private ActiveService activeService;

    @Autowired
    private TemplateService templateService;

    @Autowired
    private TaskMessageQueue<Long> taskMessageQueue;

    @Autowired
    private DataConfigService dataConfigService;

    @Autowired
    private ActiveProcessService activeProcessService;

    @Resource
    private RedisTemplate<String, TaskTypeVo> redisTemplate;

    //临时队列，后续得弄分布式，如果任务已完成，则移除，如果任务消费超时，则重新添加到任务队列 todo,
//    private Map<Long,TaskMessage<Long>> runningMap = new ConcurrentHashMap<>();

    public CrawlerTaskVo pullCrawlerTask() {

        TaskMessage<Long> taskMessage = taskMessageQueue.popTaskMessage(false);
        if (taskMessage == null) {
            return null;
        }
        return taskMessageToCrawlerTask(taskMessage);
    }


    private CrawlerTaskVo taskMessageToCrawlerTask(TaskMessage<Long> taskMessage){
        CrawlerTaskVo crawlerTaskVo = new CrawlerTaskVo();
        taskMessage.setConsumeStartTime(System.currentTimeMillis());
//        runningMap.put(taskMessage.getData(),taskMessage);

        Long activeInstanceId = taskMessage.getData();
        ActiveInstance activeInstance = this.activeInstanceService.findById(activeInstanceId);
        Active active = activeService.findById(activeInstance.getActiveId());

        Template template = templateService.findById(active.getTemplateId());

        crawlerTaskVo.setActiveInstanceId(activeInstance.getId());
        crawlerTaskVo.setActiveId(activeInstance.getActiveId());
        crawlerTaskVo.setActiveProcessId(activeInstance.getActiveProcessId());
        crawlerTaskVo.setEnableProxy(template.getEnableProxy());
        crawlerTaskVo.setEnableImage(template.getEnableImage());
        crawlerTaskVo.setEnableJavascript(template.getEnableJavascript());
        crawlerTaskVo.setEnableStyle(template.getEnableStyle());
        crawlerTaskVo.setTimeRdMs(template.getRunRdMs());
        crawlerTaskVo.setCookie(active.getCookie());

//        crawlerTaskVo.setActiveInstanceId(activeInstanceId);
        crawlerTaskVo.setAnalyseBanExpress(active.getAnalyseBanExpress());
        crawlerTaskVo.setAnalyseExpress(active.getAnalyseExpress());
        crawlerTaskVo.setAnalyseDataExpress(active.getAnalyseDataExpress());
        crawlerTaskVo.setSwitchProxyExpress(active.getSwitchProxyExpress());
        crawlerTaskVo.setUiEventScript(active.getUiEventScript());
        crawlerTaskVo.setExtractDataUrl(active.getExtractDataUrl());
        crawlerTaskVo.setSliceActive(active.getSliceActive());
        crawlerTaskVo.setAnalyseExpressType(active.getAnalyseExpressType());
        crawlerTaskVo.setVariables(active.getVariables());
        crawlerTaskVo.setUrl(activeInstance.getUrl());
        crawlerTaskVo.setWaitCssSelectorExpress(active.getWaitCssSelectorExpress());
        crawlerTaskVo.setDirtyCssSelectorExpress(active.getDirtyCssSelectorExpress());
        crawlerTaskVo.setRunTimeout(active.getRunTimeout());
        crawlerTaskVo.setStartTime(System.currentTimeMillis());

        crawlerTaskVo.setStorePageSource(template.getStorePageSource());
        return crawlerTaskVo;
    }

    public List<CrawlerTaskVo> pullMultiCrawlerTask() {
        List<TaskMessage<Long>> taskMessageList = taskMessageQueue.popTaskMultiMessage(false);
        if (CollectionUtils.isEmpty(taskMessageList)) {
            return null;
        }
        List<CrawlerTaskVo> resultList = new ArrayList<>();
        for (TaskMessage<Long> taskMessage : taskMessageList) {
            CrawlerTaskVo crawlerTaskVo = taskMessageToCrawlerTask(taskMessage);
            resultList.add(crawlerTaskVo);
        }
        return resultList;
    }
    public List<CrawlerTaskVo> pullMultiCrawlerTaskByTaskId(Long taskId) {
        List<TaskMessage<Long>> taskMessageList = taskMessageQueue.popTaskMultiMessage(taskId);
        if (CollectionUtils.isEmpty(taskMessageList)) {
           return null;
        }
        List<CrawlerTaskVo> resultList = new ArrayList<>();
        for (TaskMessage<Long> taskMessage : taskMessageList) {
            CrawlerTaskVo crawlerTaskVo = taskMessageToCrawlerTask(taskMessage);
            resultList.add(crawlerTaskVo);
        }
        return resultList;
    }

    public String getCookies(Long activeInstanceId){
        ActiveInstance activeInstance = this.activeInstanceService.findById(activeInstanceId);
        Active active = activeService.findById(activeInstance.getActiveId());
        String cookie = active.getCookie();
        return cookie;
    }

    private String [] array = null;

    public String getProxies(Long activeProcessId){
//        String url = dataConfigService.findValueByCode(CrawlerConstants.PROXIES_URL_INTERNAL);
//        按量
//        String url  = "http://v2.api.juliangip.com/postpay/getips?num=1&pt=1&result_type=text&split=1&trade_no=6093026171090145&sign=a01633d36aec2d8e652af9bd68de4b20";
        //包天
//        String url  = "http://v2.api.juliangip.com/unlimited/getips?num=1&pt=1&result_type=text&split=1&trade_no=5811406698181429&sign=f5c4e01564fd7da97717038ba464635b";
//        ResponseRes responseRes = HttpNewUtils.get(url);
//        return responseRes.getBodyText();

        //国外
//        if(array == null){
        return dataConfigService.findValueByCode(CrawlerConstants.PROXIES_TUNNEL);
//            array = proxies.split("\n");
////        }
//        int index = RandomUtils.nextInt(0, array.length);
//        return array[index];

//        String url = dataConfigService.findValueByCode(CrawlerConstants.PROXIES_URL_ABROAD);
        //按量
//        String url  = "http://v2.api.juliangip.com/postpay/getips?num=1&pt=1&result_type=text&split=1&trade_no=6093026171090145&sign=a01633d36aec2d8e652af9bd68de4b20";
        //包天
//        String url  = "http://v2.api.juliangip.com/unlimited/getips?num=1&pt=1&result_type=text&split=1&trade_no=5811406698181429&sign=f5c4e01564fd7da97717038ba464635b";
//        ResponseRes responseRes = HttpNewUtils.get(url);
//        return responseRes.getBodyText();
    }


    private void submitBan(Long activeInstanceId){
        ActiveInstance activeInstance = this.activeInstanceService.findById(activeInstanceId);
        activeProcessService.updateStatus(activeInstance.getActiveProcessId(),ProcessStatusEnum.PAUSED.getCode());
    }


    @Transactional
    public void submitCrawlerTask(CrawlerTaskSubmitParam param) {
        if(!param.isSuccess() && param.getErrorStatus() == RequestResultErrorStatusEnum.BAN.getCode()){ //如果被屏蔽，阻断拉取任务
            submitBan(param.getActiveInstanceId());
        }
        Long activeInstanceId = param.getActiveInstanceId();
        ActiveInstance activeInstance = activeInstanceService.findById(activeInstanceId);
        Active active = activeService.findById(activeInstance.getActiveId());
        if(active.getSliceActive() == YesNoEnum.YES.getCode()){  //分片
            //先往分片实例添加一一条新记录
            ActiveSliceInstance activeSliceInstance = new ActiveSliceInstance();
            activeSliceInstance.setClientTaskId(param.getClientTaskId());
            activeSliceInstance.setActiveId(active.getId());
            activeSliceInstance.setActiveInstanceId(activeInstanceId);
            activeSliceInstance.setActiveMainInstanceId(activeInstance.getActiveMainInstanceId());
            activeSliceInstance.setActiveProcessId(activeInstance.getActiveProcessId());
            activeSliceInstance.setErrContent(param.getErrContent());
            activeSliceInstance.setIntercept(false);
            activeSliceInstance.setName(activeInstance.getName());
            activeSliceInstance.setUrl(param.getUrl());
            if (StringUtils.isNotEmpty(param.getResponseBody())) {
                byte[] decode = Base64.decode(param.getResponseBody());
                activeSliceInstance.setResponseBody(new String(decode));
            }
            if (StringUtils.isNotEmpty(param.getResponseAnalyseBody())) {
                byte[] decode = Base64.decode(param.getResponseAnalyseBody());
                String str = new String(decode);
                Object obj = JSONObject.parse(str);
                activeSliceInstance.setResponseAnalyseBody(obj.toString());
            }
            activeSliceInstance.setResponseRawHeader(param.getResponseRawHeader());
            if (param.isSuccess()) {
                activeSliceInstance.setStatus(ActiveInstanceStatusEnum.SUCCESS.getCode());
            }else{
                activeSliceInstance.setStatus(ActiveInstanceStatusEnum.FAILED.getCode());
            }
            long takeTime = System.currentTimeMillis() - param.getStartTime();
            activeSliceInstance.setTakeTime(takeTime);
            activeSliceInstance.setTaskStatus(YesNoEnum.NO.getCode());
            activeSliceInstance.setTemplateId(active.getTemplateId());
            activeSliceInstanceService.save(activeSliceInstance);

            //活动实例
            activeInstance.setErrContent(param.getErrContent());

            if(param.isSuccess()){ //成功,只有成功，才写入数据
                JSONArray jsonArray = new JSONArray();
                if (StringUtils.isEmpty(activeInstance.getResponseAnalyseBody())) {
                    jsonArray.add(JSONObject.parse(activeSliceInstance.getResponseAnalyseBody()));
                }else{
                    jsonArray = (JSONArray) JSONArray.parse(activeInstance.getResponseAnalyseBody());
                    jsonArray.add(JSONObject.parse(activeSliceInstance.getResponseAnalyseBody()));
                }
                activeInstance.setResponseAnalyseBody(jsonArray.toJSONString());


            }else{ //失败
//                activeInstanceService.save(activeInstance);
            }
            activeInstanceService.save(activeInstance);

            String responseAnalyseBody = activeSliceInstance.getResponseAnalyseBody();
            //判断active 是否支持 todo
            if(StringUtils.isEmpty(responseAnalyseBody) || responseAnalyseBody.trim().equalsIgnoreCase("{}") || responseAnalyseBody.trim().equalsIgnoreCase("[]")){
                sliceFinishSubmit(param);
            }
        }else{  //非分片，一次性提交
//            TaskMessage<Long> taskMessage = runningMap.remove(param.getActiveInstanceId());
            activeInstance.setErrContent(param.getErrContent());
            if (StringUtils.isNotEmpty(param.getResponseAnalyseBody())) {
                byte[] decode = Base64.decode(param.getResponseAnalyseBody());
//                JSONObject jsonObject = new JSONObject();
                String str = new String(decode);
                Object obj= JSONObject.parse(str);
//                jsonObject.put("data", obj);
                activeInstance.setResponseAnalyseBody(obj.toString());
            }
            if (StringUtils.isNotEmpty(param.getResponseBody())) {
                byte[] decode = Base64.decode(param.getResponseBody());
                activeInstance.setResponseBody(new String(decode));
            }
            activeInstance.setResponseRawHeader(param.getResponseRawHeader());
            activeInstance.setVariables(active.getVariables());  //暂时直接替换，后续会产生中间变量
            if (param.isSuccess()) {
                activeInstance.setStatus(ActiveInstanceStatusEnum.SUCCESS.getCode());
            } else {

                activeInstance.setStatus(ActiveInstanceStatusEnum.FAILED.getCode());
            }
            long takeTime = System.currentTimeMillis() - param.getStartTime();
            activeInstance.setTakeTime(takeTime);
            try {
                //任务结束，设置为0
                activeInstance.setTaskStatus(YesNoEnum.NO.getCode());
                //回写数据库
                activeInstanceService.activeRequestUpdate(activeInstance);
//                runningMap.remove(activeInstanceId);  //已经移除进行中的队列，要先于 addTaskMessage

//                if(!param.isSuccess()){
//                    //失败,重新添加到队列
//                    taskMessageQueue.addTaskMessage(taskMessage);
//                }
            } catch (Exception e) {
                logger.error("submitCrawlerTask activeInstance failed", e);
            }
        }
    }


    @Transactional
    public void sliceFinishSubmit(CrawlerTaskSubmitParam param) {
//        TaskMessage<Long> taskMessage = runningMap.remove(param.getActiveInstanceId());
        Long activeInstanceId = param.getActiveInstanceId();
        ActiveInstance activeInstance = activeInstanceService.findById(activeInstanceId);
//        Active active = activeService.findById(activeInstance.getActiveId());
        activeInstance.setErrContent(param.getErrContent());
        //先判断对应的客户端task中有没有失败的，如果有失败，则整个失败，否则成功 todo
        Integer status = param.isSuccess() ? ActiveInstanceStatusEnum.SUCCESS.getCode() : ActiveInstanceStatusEnum.FAILED.getCode();
        activeInstance.setStatus(status);
        long takeTime = System.currentTimeMillis() - param.getStartTime();
        activeInstance.setTakeTime(takeTime);
        try {
            //任务结束，设置为0
            activeInstance.setTaskStatus(YesNoEnum.NO.getCode());
            //回写数据库
            activeInstanceService.activeRequestUpdate(activeInstance);
//            runningMap.remove(activeInstanceId);  //已经移除进行中的队列，要先于 addTaskMessage

//            if(!param.isSuccess()){
//                //失败,重新添加到队列
//                taskMessageQueue.addTaskMessage(taskMessage);
//            }
        } catch (Exception e) {
            logger.error("submitCrawlerTask activeInstance failed", e);
        }
    }


}
